/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef ZK_ADAPTOR_H_
#define ZK_ADAPTOR_H_
#include "zookeeper.jute.h"
#ifdef THREADED
#include <linux/wait.h>
#include <linux/spinlock.h>
#endif
#include "zookeeper.h"
#include "shard.h"
#include "block_storage.h"
#include <linux/socket.h>
#include <linux/net.h>

/* predefined xid's values recognized as special by the server */
#define WATCHER_EVENT_XID -1 
#define PING_XID -2
#define AUTH_XID -4
#define SET_WATCHES_XID -8

/* zookeeper state constants */
#define EXPIRED_SESSION_STATE_DEF -112
#define AUTH_FAILED_STATE_DEF -113
#define CONNECTING_STATE_DEF 1
#define ASSOCIATING_STATE_DEF 2
#define CONNECTED_STATE_DEF 3

/* zookeeper event type constants */
#define CREATED_EVENT_DEF 1
#define DELETED_EVENT_DEF 2
#define CHANGED_EVENT_DEF 3
#define CHILD_EVENT_DEF 4
#define SESSION_EVENT_DEF -1
#define NOTWATCHING_EVENT_DEF -2


#ifdef __cplusplus
extern "C" {
#endif

struct _buffer_list;
struct _completion_list;

typedef struct _buffer_head {
    struct _buffer_list *volatile head;
    struct _buffer_list *last;
#ifdef THREADED
    spinlock_t lock;
    unsigned long flags;
#endif
} buffer_head_t;

typedef struct _completion_head {
    struct _completion_list *volatile head;
    struct _completion_list *last;
#ifdef THREADED
    wait_queue_head_t cond;
#endif
} completion_head_t;

void lock_buffer_list(buffer_head_t *l);
void unlock_buffer_list(buffer_head_t *l);
void lock_completion_list(completion_head_t *l);
void unlock_completion_list(completion_head_t *l);


/**
 * This structure represents a packet being read or written.
 */
typedef struct _buffer_list {
    char *buffer;
    int len; /* This represents the length of sizeof(header) + length of buffer */
    int curr_offset; /* This is the offset into the header followed by offset into the buffer */
    struct _buffer_list *next;
} buffer_list_t;

/* the size of connect request */
#define HANDSHAKE_REQ_SIZE 44
/* connect request */
struct connect_req {
    int64_t lastZxidSeen;
    int32_t timeOut;
    int64_t sessionId;
};

/* the connect response */
struct prime_struct {
    int32_t len;
    int32_t timeOut;
    int64_t sessionId;
    int32_t cxid;
}; 

#ifdef THREADED
/* this is used by mt_adaptor internally for thread management */
struct adaptor_threads {
     struct task_struct *io;
     struct task_struct *completion;
     int threadsToWait;         // barrier
     wait_queue_head_t cond;       // barrier's conditional
     spinlock_t zh_lock;   // critical section lock
     unsigned long flags; // flags for the lock
     int self_pipe[2];
     struct file* self_pipe_file[2];
};
#endif


/**
 * This structure represents the connection to zookeeper.
 */

struct _zhandle {
    void *storage;
    int fd[SERVER_COUNT]; /* the descriptor used to talk to zookeeper */
    char *hostname; /* the hostname of zookeeper */
    struct sockaddr_storage *addrs; /* the addresses that correspond to the hostname */
    int addrs_count; /* The number of addresses in the addrs array */
    struct timeval last_recv[SERVER_COUNT]; /* The time that the last message was received */
    struct timeval last_send[SERVER_COUNT]; /* The time that the last message was sent */
    struct timeval last_ping[SERVER_COUNT]; /* The time that the last PING was sent */
    struct timeval next_deadline[SERVER_COUNT]; /* The time of the next deadline */
    struct timeval next_connect[SERVER_COUNT];
    int recv_timeout; /* The maximum amount of time that can go by without 
     receiving anything from the zookeeper server */
    buffer_list_t *input_buffer[SERVER_COUNT]; /* the current buffer being read in */
    buffer_head_t to_process[SERVER_COUNT]; /* The buffers that have been read and are ready to be processed. */
    buffer_head_t to_send[SERVER_COUNT]; /* The packets queued to send */
    clientid_t client_id;
    long long last_zxid;
    struct _buffer_list primer_buffer[SERVER_COUNT]; /* The buffer used for the handshake at the start of a connection */
    struct prime_struct primer_storage[SERVER_COUNT]; /* the connect response */
    char primer_storage_buffer[SERVER_COUNT][20]; /* the true size of primer_storage */
    volatile int state[SERVER_COUNT];
    void *context;
    /* zookeeper_close is not reentrant because it de-allocates the zhandler. 
     * This guard variable is used to defer the destruction of zhandle till 
     * right before top-level API call returns to the caller */
    int32_t ref_counter;
    volatile int close_requested;
    void *adaptor_priv;
    /* Used for debugging only: non-zero value indicates the time when the zookeeper_process
     * call returned while there was at least one unprocessed server response 
     * available in the socket recv buffer */
    struct timeval socket_readable[SERVER_COUNT];
    
};


int adaptor_init(zhandle_t *zh);
void adaptor_finish(zhandle_t *zh);
void adaptor_destroy(zhandle_t *zh);
int adaptor_send_queue(zhandle_t *zh, int timeout, int sid);
int process_async(int outstanding_sync);
void process_completions(zhandle_t *zh);
int flush_send_queue(zhandle_t*zh, int timeout, int sid);
char* sub_string(zhandle_t *zh, const char* server_path);
void free_duplicate_path(const char* free_path, const char* path);
void zoo_lock_auth(zhandle_t *zh);
void zoo_unlock_auth(zhandle_t *zh);

// critical section guards
void enter_critical(zhandle_t* zh);
void leave_critical(zhandle_t* zh);
// zhandle object reference counting
void api_prolog(zhandle_t* zh);
int api_epilog(zhandle_t *zh, int rc);
// returns the new value of the ref counter
int32_t inc_ref_counter(zhandle_t* zh,int i);

#ifdef THREADED
// atomic post-increment
int32_t fetch_and_add(volatile int32_t* operand, int incr);
// in mt mode process session event asynchronously by the completion thread
#define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
#else
// in single-threaded mode process session event immediately
//#define PROCESS_SESSION_EVENT(zh,newstate) deliverWatchers(zh,ZOO_SESSION_EVENT,newstate,0)
#define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
#endif

#ifdef __cplusplus
}
#endif

#endif /*ZK_ADAPTOR_H_*/


